package com.wuwangfu.window.event;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @Author jcshen
 * @Date 2023-02-27
 * @PackageName:com.wuwangfu.window.event
 * @ClassName: EventTimeTumbleWindow2s
 * @Description:
 * @Version 1.0.0
 *
 * 先keyBy，再按照EventTime划分滚动窗口
 * 设置延迟时间为 2秒
 *
 * https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/windows/#tumbling-windows
 */
public class EventTimeTumbleWindow2s {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        /**
         * 1000,spark,1
         * 3333,hive,1
         * 4999,spark,2
         * 4000,hive,2
         * 5555,flink,3
         * 6999,hive,2
         *
         */
        DataStreamSource<String> line = env.socketTextStream("localhost", 8888);
        //生成watermark
        //当前分区中携带的 最大的EventTime - 乱序延迟时间 >= 窗口结束时间 就会触发该窗口
        SingleOutputStreamOperator<String> dataWithWatermark = line.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.parseLong(element.split(",")[0]);
            }
        });
        //
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataWithWatermark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] fields = value.split(",");
                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });
        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);
        //开窗
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));
        //聚合
        windowed.sum(1).print();


        env.execute();
    }
}
